package com.key.w8.http1.connection.pool;

import com.key.w8.common.execute.ChainExecuteHandler;
import com.key.w8.http1.connection.ChannelBuilder;
import com.key.w8.http1.connection.Connection;
import com.key.w8.http1.connection.connectionImpl.ClientHttp1PersistentConnection;
import com.key.w8.http1.connection.ConnectionBuilder;
import com.key.w8.http1.connection.Http1ConnectionInitializer;
import com.key.w8.http1.connection.environment.ClientPoolEnvironment;
import com.key.w8.http1.connection.connectionImpl.DefaultHttp1Connection;
import com.key.w8.http1.constant.TSLProtocol;
import com.key.w8.http1.handler.Http1FullHandlerBuilder;
import com.key.w8.http1.handler.codec.CodecBuilder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import javax.swing.plaf.PanelUI;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static com.key.w8.http1.constant.TSLProtocol.TSL_V2;

/**
 * @Author: k8
 * @CreateTime: 2025-03-03
 * @Version: 1.0
 */
@Slf4j
public class ClientConnectionBuilder implements ConnectionBuilder<ClientHttp1PersistentConnection,ConnectionPool.ConnectionWorker>, ChannelBuilder {
    private final ClientPoolEnvironment environment;
    private final EventLoopGroup group;
    private boolean isServer;
    private  Http1FullHandlerBuilder handlerBuilder;
    private final List<ChainExecuteHandler> extendExecuteHandlers = new LinkedList<>();
    private final List<ChainExecuteHandler> extendErrorHandlers = new LinkedList<>();
    private final List<Connection.CloseListener> closeListeners = new LinkedList<>();
    private final List<Connection.ErrorListener> errorListeners = new LinkedList<>();
    private final List<Connection.CreatedListener> createdListeners = new LinkedList<>();
    private final List<Connection.WriteRequestListener> writeRequestListeners = new LinkedList<>();
    public static final boolean DEFAULT_SSL_ENABLE = false;  // 默认不启用SSL
    private boolean sslEnable = DEFAULT_SSL_ENABLE;           // 是否启用 SSL
    public static final String DEFAULT_HOST = "localhost";
    public static final int DEFAULT_PORT = 4399;
    private int port = DEFAULT_PORT;
    private String tslProtocol = TSL_V2;
    private String host = DEFAULT_HOST;
    public ClientConnectionBuilder setHost(String host){
        this.host = host;
        return this;
    }
    public ClientConnectionBuilder setPort(int port){
        this.port = port;
        return this;
    }
    public ClientConnectionBuilder setSslEnable(boolean sslEnable){
        this.sslEnable = sslEnable;
        return  this;
    }
    public ClientConnectionBuilder setTslProtocol(String tslProtocol){
        this.tslProtocol = tslProtocol;
        return this;
    }
    @Override
    public void addExecuteHandler(ChainExecuteHandler handler) {
        assert handler != null;
        extendExecuteHandlers.add(handler);
    }

    @Override
    public void addExecuteErrorHandler(ChainExecuteHandler handler) {
        assert handler != null;
        extendErrorHandlers.add(handler);
    }

    @Override
    public void addErrorListener(Connection.ErrorListener errorListener) {
        assert errorListener != null;
        errorListeners.add(errorListener);
    }

    @Override
    public void addCloseListener(Connection.CloseListener closeListener) {
        assert closeListener != null;
        closeListeners.add(closeListener);
    }

    @Override
    public void addCreatedListener(Connection.CreatedListener createdListener) {
        assert createdListener != null;
        createdListeners.add(createdListener);
    }

    @Override
    public void addWriteRequestListener(Connection.WriteRequestListener writeRequestListener) {
        assert writeRequestListener != null;
        writeRequestListeners.add(writeRequestListener);
    }

    /**
     * netty默认的负债均衡也是通过一个atomicInteger进行轮询选择
     * 可以自定义，但是灵活性不够
     * 如果直接轮询，当有连接关闭后又创建
     * 容易导致不同的连接被注册到同一个eventLoop上
     * 所以使用单线程group，创建多个，自定义轮询
     */

    public ClientConnectionBuilder(ClientPoolEnvironment environment, CodecBuilder codecBuilder) {
        assert environment != null;
        this.environment = environment;
        this.group = new NioEventLoopGroup(environment.getMaxThreads());
        this.isServer = environment.isServer();
        Http1FullHandlerBuilder handlerBuilder = new Http1FullHandlerBuilder(codecBuilder,environment.isKeepAlive());
        handlerBuilder
                .setServer(environment.isServer())
                .setHeartbeatInterval(environment.getHeartbeatInterval())
                .setConnectionTimeout(environment.getConnectionTimeout())
                .setMaxContentLength(environment.getMaxContentLength())
                .setCompressResponseContent(environment.isCompressResponseContent());
        this.handlerBuilder = handlerBuilder;
    }
    protected ClientHttp1PersistentConnection getConnection(ClientPoolEnvironment environment, EventLoopGroup group){
        return new ClientHttp1PersistentConnection(environment.getResponseTimeOut(),TimeUnit.MILLISECONDS);
    }
    @Override
    public ClientHttp1PersistentConnection build(String host,int port){
        if (isServer) throw new IllegalStateException("This builder is server");
        ClientHttp1PersistentConnection connection = getConnection(environment,group);
        this.extendExecuteHandlers.forEach(connection::addChainExecuteHandler);
        Bootstrap bootstrap = doBuildStrap(group,connection);
        postBuild(connection);
        AtomicInteger reConnectTimes;
        if (environment.isReConnect()){
            reConnectTimes = new AtomicInteger(environment.getReConnectTimes());
        } else {
            reConnectTimes = null;
        }
        try {
            doConnectSync(bootstrap,connection,reConnectTimes,host,port);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return connection;
    }
    @Override
    public ClientHttp1PersistentConnection build(){
        if (isServer) throw new IllegalStateException("This builder is server");
        ClientHttp1PersistentConnection connection = getConnection(environment,group);
        this.extendExecuteHandlers.forEach(connection::addChainExecuteHandler);
        Bootstrap bootstrap = doBuildStrap(group,connection);
        postBuild(connection);
        AtomicInteger reConnectTimes;
        if (environment.isReConnect()){
            reConnectTimes = new AtomicInteger(environment.getReConnectTimes());
        } else {
            reConnectTimes = null;
        }
        try {
            doConnectSync(bootstrap,connection,reConnectTimes);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return connection;
    }
    @Override
    public ClientHttp1PersistentConnection build(ConnectionPool.ConnectionWorker connectionWorker){
        if (isServer) throw new IllegalStateException("This builder is server");
        ClientHttp1PersistentConnection connection = getConnection(environment,group);
        this.extendExecuteHandlers.forEach(connection::addChainExecuteHandler);
        Bootstrap bootstrap = doBuildStrap(group,connection);
        connection.addCreatedListener((connection1)->{
            connectionWorker.start();
        });
        postBuild(connection);
        AtomicInteger reConnectTimes;
        if (environment.isReConnect()){
            reConnectTimes = new AtomicInteger(environment.getReConnectTimes());
        } else {
            reConnectTimes = null;
        }
        connectionWorker.setConnectListener(()->{
            doConnect(bootstrap,connection,reConnectTimes);
        });
        connectionWorker.setConnection(connection);

        return connection;
    }
    private void postBuild(Connection connection){
        this.closeListeners.forEach(connection::addCloseListener);
        this.createdListeners.forEach(connection::addCreatedListener);
        this.errorListeners.forEach(connection::addErrorListener);
        this.extendErrorHandlers.forEach(connection::addChainErrorHandler);
        this.writeRequestListeners.forEach(connection::addWriteRequestListener);
    }
    @Override
    public Bootstrap doBuildStrap(EventLoopGroup group, DefaultHttp1Connection connection){
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)  // 连接超时5秒‌
                .option(ChannelOption.TCP_NODELAY, false)  // 禁用Nagle算法‌
                .option(ChannelOption.SO_KEEPALIVE, true)  // 启用TCP保活‌
                .handler(new Http1ConnectionInitializer(connection,handlerBuilder,sslEnable,tslProtocol,isServer,environment.getCrtPath(),environment.getPrivateKeyPath()));
        return bootstrap;
    }
    @Override
    public void doConnect(Bootstrap bootstrap, DefaultHttp1Connection connection, AtomicInteger reConnectTimes){
        ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress(host, port));
                channelFuture.addListener(future -> {
                    log.debug("-------------------------connect.");
                    try{
                        future.get();
                    }catch (Throwable throwable){
                        log.debug("connect error");
                    }
                    if (future.isSuccess()){
                        log.debug("-----------------------connect success.");
                    }else if (environment.isReConnect() && reConnectTimes != null && reConnectTimes.get() > 0){
                            reConnectTimes.decrementAndGet();
                            doConnect(bootstrap,connection,reConnectTimes);
                    }else {
                        connection.close();
                    }
                });
    }

    public void doConnectSync(Bootstrap bootstrap, DefaultHttp1Connection connection, AtomicInteger reConnectTimes) throws IOException {
        doConnectSync(bootstrap,connection,reConnectTimes,host,port);
    }
    public void doConnectSync(Bootstrap bootstrap, DefaultHttp1Connection connection, AtomicInteger reConnectTimes,String host,int port) throws IOException {
        try {
            log.debug("-------------------------connect.");
            ChannelFuture channelFuture = null;
            channelFuture = bootstrap.connect(new InetSocketAddress(host, port)).sync(); // 阻塞等待连接完成
            if (channelFuture.isSuccess()) {
                log.debug("-----------------------connect success.");
                // 连接成功，可以做绑定等操作
            } else {
                log.debug("connect failed.");
                if (environment.isReConnect() && reConnectTimes != null && reConnectTimes.get() > 0) {
                    reConnectTimes.decrementAndGet();
                    doConnectSync(bootstrap, connection, reConnectTimes,host,port); // 重试连接
                } else {
                    connection.close();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("connect interrupted.", e);
            connection.close();
        } catch (Exception e) {
            log.error("connect error", e);
            if (environment.isReConnect() && reConnectTimes != null && reConnectTimes.get() > 0) {
                reConnectTimes.decrementAndGet();
                doConnectSync(bootstrap, connection, reConnectTimes,host,port); // 重试连接
            } else {
                connection.close();
            }
        }
    }
}
